//package com.open.source.flink.cdc;
//
//import com.ververica.cdc.connectors.mysql.source.MySqlSource;
//import com.ververica.cdc.connectors.mysql.table.StartupOptions;
//import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
//import org.apache.flink.api.common.eventtime.WatermarkStrategy;
//import org.apache.flink.api.common.restartstrategy.RestartStrategies;
//import org.apache.flink.runtime.state.filesystem.FsStateBackend;
//import org.apache.flink.streaming.api.CheckpointingMode;
//import org.apache.flink.streaming.api.datastream.DataStreamSource;
//import org.apache.flink.streaming.api.environment.CheckpointConfig;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//
///**
// * @author ZonLen since on 2022/7/20 下午11:04
// */
//public class FlinkCDC1 {
//  public static void main(String[] args) throws Exception {
//    //1.创建执行环境
//    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//    //2.设置并行度
//    env.setParallelism(1);
//
//    //3.开启 checkpoint，并设置相关参数
//    env.enableCheckpointing(5000);
//    //设置 CK 一致性语义
//    env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//    //设置任务关闭的时候保留最后一次 CK 数据
//    env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//    //设置从 CK 自动重启策略
//    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));
//    //设置状态后端
//    env.setStateBackend(new FsStateBackend("hdfs://ns1/flinkcdc/checkpoint2",true));
//
//    //设置访问 HDFS 的用户名
//    System.setProperty("HADOOP_USER_NAME", "hadoop");
//
//    //4.创建 FlinkCDC 的 Source
//    MySqlSource<String> build =
//        //DebeziumSourceFunction<String> build =
//        MySqlSource.<String>builder()
//            .hostname("hadoop3")
//            .port(3306)
//            .username("root")
//            .password("qwert123")
//            .databaseList("test")
//            //可选项配置，如果不指定参数，则会读取上一个配置的所有表数据
//            //注意：指定时需要使用：db.table 的方式
//            .tableList("test.t_user")
//            .startupOptions(StartupOptions.initial())
//            .deserializer(new StringDebeziumDeserializationSchema())
//            .build();
//
//
//    //5.source:使用 CDC Source 从 mysql 中读取数据
//    DataStreamSource<String> source = env.fromSource(build, WatermarkStrategy.noWatermarks(), "flinkcdc-mysql");
//    //DataStreamSource<String> source = env.addSource(build);
//
//    //6.sink:打印数据
//    source.print();
//
//    env.execute("Flink CDC");
//
//  }
//}
//
